package com.example.demo.hadoop.mapreduce;

import com.example.demo.hadoop.reduce.model.GroupSortModel;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

public class GroupSort extends Configured implements Tool {

    public static class GroupSortMapper extends Mapper<LongWritable, Text, GroupSortModel, IntWritable> {
        private static final GroupSortModel groupSortModel = new GroupSortModel();
        private static final IntWritable num = new IntWritable();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] split = line.split("\t");
            if (split != null && split.length >= 2) {
                groupSortModel.set(Integer.parseInt(split[0]), Integer.parseInt(split[1]));
                num.set(Integer.parseInt(split[1]));
                context.write(groupSortModel, num);
            }
        }
    }

    public static class GroupSortPartitioner extends Partitioner<GroupSortModel, IntWritable> {

        @Override
        public int getPartition(GroupSortModel groupSortModel, IntWritable intWritable, int i) {
            return Math.abs(groupSortModel.getName() * 127) % i;
        }
    }

    public static class GroupSortComparator extends WritableComparator {
        public GroupSortComparator() {
            super(GroupSortModel.class, true);
        }

        @Override
        public int compare(WritableComparable a, WritableComparable b) {
            GroupSortModel mode1 = (GroupSortModel) a;
            GroupSortModel mode2 = (GroupSortModel) b;
            int num1 = mode1.getNum();
            int num2 = mode2.getNum();
            return num1 == num2 ? 0 : (num1 < num2 ? -1 : 1);
        }
    }

    public static class GroupSortReduce extends Reducer<GroupSortModel, IntWritable, Text, IntWritable> {
        private static final Text name = new Text();

        @Override
        protected void reduce(GroupSortModel key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            name.set(key + "");
            for (IntWritable value : values) {
                System.out.println("reduce输出: " + key + " " + value);
                context.write(name, value);
            }
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        String path = args[1];
        Configuration conf = new Configuration();
        // 如果目标文件存在则删除

        Path outPath = new Path(path);
        FileSystem fs = outPath.getFileSystem(conf);
        if (fs.exists(outPath)) {
            boolean flag = fs.delete(outPath, true);
        }
        Job job = Job.getInstance(conf, "groupSort");
        job.setJarByClass(GroupSort.class);
        job.setMapperClass(GroupSort.GroupSortMapper.class);
        job.setReducerClass(GroupSort.GroupSortReduce.class);
        // 设置mapper和reduce的输出格式，如果相同则只需设置一个
        job.setOutputKeyClass(GroupSortModel.class);
        job.setOutputValueClass(NullWritable.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception{
        String[] filePath = {
                "hdfs://192.168.51.3:9000/java/groupSort.txt",
                "hdfs://192.168.51.3:9000/output/groupSort"
        };
        int ec = ToolRunner.run(new Configuration(), new GroupSort(), filePath);
    }
}
